package co.sugon.log.searchEngineUtils;

import co.sugon.log.utils.ResourceLoader;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.*;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;


public class SparkJdbcUtil {


	private final Logger logger = LoggerFactory.getLogger(SparkJdbcUtil.class);

	private final String driverName = "org.apache.hive.jdbc.HiveDriver";

	private final String hiveThriftUrl = "jdbc:hive2://ip:port/";

	private static SparkJdbcUtil util;

	private static String username = "root";

	private static String password = "";

	private String url; // 使用中的url

	private List<String> urls;

	private SparkJdbcUtil() throws ClassNotFoundException {
		String username = ResourceLoader.getExtraProperties("spark.thrift.username");
		String password = ResourceLoader.getExtraProperties("spark.thrift.password");
		String port = ResourceLoader.getExtraProperties("spark.thrift.port");
		String ip = ResourceLoader.getExtraProperties("spark.thrift.ip");
		if (StringUtils.isBlank(port) || StringUtils.isBlank(ip)) {
			logger.error("获取Spark Thrift Ip 或 Port为空!");
			throw new RuntimeException("Spark Thrift Ip or Port is Empty!");
		}
		urls = Lists.newArrayList();
		List<String> ips = Arrays.asList(ip.replaceAll(" ", "").trim()
				.split(","));
		for (String ipLine : ips) {
			String hiveUrl = hiveThriftUrl;
			if (ipLine.contains(";")) {
				hiveUrl = hiveUrl.replace("ip:port", url);
			} else {
				hiveUrl = hiveUrl.replace("ip", ipLine).replace("port", port);
			}
			urls.add(hiveUrl);
		}
		if (ips.size() != 0) {
			url = urls.get(0);
		}
		SparkJdbcUtil.username = username;
		SparkJdbcUtil.password = password;
		Class.forName(driverName);
	}

	/**
	 * Spark Thrift Jdbc实例
	 * 
	 * @return
	 * @throws ClassNotFoundException
	 * @throws SQLException
	 */
	public static SparkJdbcUtil getInstance() throws ClassNotFoundException,
			SQLException {
		if (util == null) {
			synchronized (SparkJdbcUtil.class) {
				util = new SparkJdbcUtil();
			}
		}
		return util;
	}

	/**
	 * 获取Spark Thrift连接,如果当前URL不可用,轮询所有URL找到可用连接
	 * 
	 * @return
	 * @throws Exception
	 */
	public Connection getConnection() throws Exception {
		Connection con = null;
		String usingUrl = url;
		for (String urlTemp : urls) {
			try {
				con = DriverManager.getConnection(usingUrl, username, password);
				// 如果现在默认使用的URL和可用的URL不相同,修改当前可用URL
				if (url == null || !url.equals(usingUrl)) {
					synchronized (SparkJdbcUtil.class) {
						url = usingUrl;
					}
				}
				logger.info("当前使用Spark Thrift地址为" + url);
				return con;
			} catch (Exception e) {
				logger.error("Spark Thrift连接失败,IP地址为：" + url);
				logger.error("开始使用" + urlTemp + "重试!");
				usingUrl = urlTemp;
			}
		}
		logger.info("所有Spark Thrift连接都不可用,请检查配置文件!");
		return con;
	}

	private void closeConnection(Connection conn) {
		if (conn == null)
			return;
		try {
			if (!conn.isClosed()) {
				// 关闭数据库连接
				conn.close();
			}
		} catch (SQLException e) {
			logger.info("#ERROR-SparkJdbc# :关闭数据库连接发生异常，请检查！"+e.getMessage());
		}
	}

	/**
	 * 查询sql语句,返回键值类型
	 * 
	 * @param sql
	 * @return
	 * @throws Exception
	 */
	public List<Map<String, Object>> querySqlForKeyValue(String sql)
			throws Exception {
		Connection con = getConnection();
		Statement stmt = null;
		ResultSet rs = null;
		List<Map<String, Object>> list = new ArrayList<Map<String, Object>>();
		try {
			stmt = con.createStatement();
			rs = stmt.executeQuery(sql);
			ResultSetMetaData md = rs.getMetaData();
			int columnCount = md.getColumnCount();
			while (rs.next()) {
				Map<String, Object> map = Maps.newHashMap();
				for (int i = 1; i <= columnCount; i++) {
					map.put(md.getColumnName(i), rs.getObject(i));
				}
				list.add(map);
			}
		} catch (SQLException ex) {
			logger.error("查询sql失败:" + sql+"   |  "+ex.getMessage());
			throw new SQLException("查询sql失败:" + sql);
		} finally {
			if (rs != null) {
				rs.close();
			}
			if (stmt != null) {
				stmt.close();
			}
			if (con != null) {
				closeConnection(con);
			}
		}
		return list;
	}

	/**
	 * 执行sql语句
	 * 
	 * @param sql
	 * @return
	 * @throws Exception
	 */
	public boolean excuteSql(String sql) throws Exception {
		Connection con = getConnection();
		boolean flag = false;
		Statement stmt = null;
		try {
			stmt = con.createStatement();
			flag = stmt.execute(sql);
		} catch (SQLException e) {
			logger.error("执行sql失败:" + sql+"    |    " + e.getMessage());
			throw new SQLException("执行sql失败:" + sql);
		} finally {
			if (stmt != null) {
				stmt.close();
			}
			if (con != null) {
				closeConnection(con);
			}
		}
		return flag;
	}

	/**
	 * 查询sql语句,返回List<String[]>
	 * 
	 * String[]为一列数据
	 * 
	 * @param sql
	 * @return
	 * @throws Exception
	 */
	public List<Object[]> queryRows(String sql) throws Exception {
		Connection con = getConnection();
		Statement stmt = null;
		ResultSet rs = null;
		List<Object[]> list = new ArrayList<Object[]>();
		try {
			stmt = con.createStatement();
			rs = stmt.executeQuery(sql);
			ResultSetMetaData md = rs.getMetaData();
			int columnCount = md.getColumnCount();
			while (rs.next()) {
				Object[] row = new Object[columnCount];
				for (int i = 0; i < columnCount; i++) {
					row[i] = rs.getObject(i + 1);
				}
				list.add(row);
			}
			logger.info("表列信息:" + list.toString());
		} catch (SQLException ex) {
			logger.error("查询sql失败:" + sql+"    |    " +ex);
			throw new SQLException("查询sql失败:" + sql);
		} finally {
			if (rs != null) {
				rs.close();
			}
			if (stmt != null) {
				stmt.close();
			}
			if (con != null) {
				closeConnection(con);
			}
		}
		return list;
	}

	/**
	 * 将结果街转化为json 字符串
	 */
	public String excuteQueryForJson(String sql) throws Exception {
		Connection con = getConnection();
		boolean flag = false;
		Statement stmt = null;
		try {
			stmt = con.createStatement();
			ResultSet rs = stmt.executeQuery(sql);
			JSONArray array = new JSONArray();  
			   // 获取列数  
			   ResultSetMetaData metaData = rs.getMetaData();  
			   int columnCount = metaData.getColumnCount();  
			    
			   // 遍历ResultSet中的每条数据  
			    while (rs.next()) {  
			        JSONObject jsonObj = new JSONObject();  
			         
			        // 遍历每一列  
			        for (int i = 1; i <= columnCount; i++) {  
			            String columnName =metaData.getColumnLabel(i);  
			            String value = rs.getString(columnName);  
			            jsonObj.put(columnName, value);  
			        }
			        array.add(jsonObj); 
			    }
			    String json = array.toJSONString();
			   return json;
		} catch (SQLException e) {
			logger.error("执行sql失败:" + sql+"  |   "+e.getMessage());
			throw new SQLException("执行sql失败:" + sql);
		} finally {
			if (stmt != null) {
				stmt.close();
			}
			if (con != null) {
				closeConnection(con);
			}
		}
	}

	/**
	 * 执行多条sql语句
	 * 
	 * @param sqls
	 * @return
	 * @throws Exception
	 */
	public boolean excuteMultSql(String[] sqls) throws Exception {
		Connection con = getConnection();
		boolean flag = false;
		Statement stmt = null;
		String sqlTemp = null;
		try {
			stmt = con.createStatement();
			if (sqls != null) {
				for (String sql : sqls) {
					sqlTemp = sql;
					flag = stmt.execute(sqlTemp);
				}
			}
		} catch (SQLException e) {
			logger.error("执行sql失败:" + sqls+"  |   "+e.getMessage());
			throw new SQLException("执行sql失败:" + sqlTemp);
		} finally {
			if (stmt != null) {
				stmt.close();
			}
			if (con != null) {
				closeConnection(con);
			}
		}
		return flag;
	}


	public long countQuery(String sql) throws Exception {
		Connection con = getConnection();
		boolean flag = true;
		Statement stmt = null;
		long count=0l;
		try {
			stmt = con.createStatement();
			ResultSet rs=null;
			rs=stmt.executeQuery(sql);

			while(rs.next()) {
				count=rs.getInt(1);
			}

		} catch (SQLException e) {
			logger.error("执行sql失败: " + sql, e);
			throw new SQLException("执行sql失败:" + sql);
		} finally {
			if (stmt != null) {
				stmt.close();
			}
			if (con != null) {
				closeConnection(con);
			}
		}
		return count;
	}

}
